package org.ykhl.sdk.appserver.xixi;

import org.ykhl.mq.commons.BaseMessage;
import org.ykhl.mq.commons.ConnectionException;
import org.ykhl.mq.commons.Constants;
import org.ykhl.sdk.appserver.MessageQueueFactoryKafkaImpl;
import org.ykhl.sdk.appserver.MessageQueueKafkaImpl;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by zkf on 2016/4/19.
 */
public class MultipleResponsesTestlww {
    public static void main(String[] args) throws InterruptedException, ConnectionException {

//        MessageQueueKafkaImpl messageQueue = MessageQueueFactoryKafkaImpl.newInstance();
        MessageQueueKafkaImpl messageQueue = MessageQueueFactoryKafkaImpl.newInstanceWithSecurityFilePath(
                Constants.CLASSPATH + "security-files/xixi/lww/client.keystoretestapplww.jks");
        messageQueue.setServerAddress(Constants.KAFKA_SERVER_LIST_DEV);
        messageQueue.setListener(MessageQueueKafkaImpl.NO_OP_LISTENER);
        messageQueue.setUseSsl(false);
        messageQueue.start();

        try {
            int cntExpexted = 110;
            String fid = "BS35006";
            String request = "<ESBEntry><AccessControl><UserName>dlzs</UserName><Password>1234</Password><Fid>BS35006</Fid></AccessControl><MessageHeader><Fid>BS35006</Fid><SourceSysCode>S16</SourceSysCode><TargetSysCode>S01</TargetSysCode><MsgDate>2016-04-19 11:30:45</MsgDate></MessageHeader><MsgInfo><onceFlag>0</onceFlag><startNum></startNum><endNum></endNum><Msg></Msg><query item=\"Register_Plan_Date\" compy=\"&gt;=\" value=\"'20160419000000'\" splice=\"and\"/><query item=\"Register_Plan_Date\" compy=\"&lt;=\" value=\"'20160420000000'\" splice=\"and\"/><query item=\"Visit_Status\" compy=\"=\" value=\"'1'\" splice=\"and\"/></MsgInfo></ESBEntry>";
            String ruleName = "app";
            BlockingQueue<BaseMessage> queue = messageQueue.sendRequest(
                    fid, "5.0", request, "devSign","");

            List<BaseMessage> array = new LinkedList<>();
            while (true) {
                BaseMessage msg = queue.poll(60, TimeUnit.SECONDS);
                if (msg == null)
                    break;

                array.add(msg);
                System.out.println("ssss" + msg.getMessageBody());
                System.out.println("cur cnt : " + array.size());
                if (array.size() >= cntExpexted)
                    break;
            }

            System.out.println("total msg Cnt: " + array.size());
            System.out.println("last msg content: " + array.get(array.size() - 1));
        } finally {
            messageQueue.stop();
        }
    }
}
